package org.pupil.socket.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultPromise;
import org.pupil.common.RpcFuture;
import org.pupil.common.RpcRequestHolder;
import org.pupil.common.constants.MsgType;
import org.pupil.common.constants.ProtocolConstants;
import org.pupil.common.constants.RpcSerialization;
import org.pupil.service.HelloService;
import org.pupil.socket.codec.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.util.concurrent.TimeUnit;

/**
 * @Description: 服务发送端
 * @Author: pupil
 * @Date: 2024/07/09 下午 4:00
 */
public class Client {

    private Logger logger = LoggerFactory.getLogger(Client.class);

    private final String host;

    private final Integer port;

    private final Bootstrap bootstrap;
    private final EventLoopGroup eventLoopGroup;

    private final ChannelFuture channelFuture;

    public Client(String host, Integer port) throws InterruptedException {
        this.host = host;
        this.port = port;

        bootstrap = new Bootstrap();
        eventLoopGroup = new NioEventLoopGroup(4);

        bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline()
                                .addLast(new RpcEncoder())
                                .addLast(new RpcDecoder())
                                .addLast(new ClientHandler());
                    }
                });
        // 阻塞等待连接，连接成功后返回异步监听器用于通信
        channelFuture = bootstrap.connect(host, port).sync();
    }

    // 发送请求
    public void sendRequest(Object request) {
        // 获取异步监听器的socket(套接字)发送请求
        channelFuture.channel().writeAndFlush(request);
    }

    public static void main(String[] args) throws Exception {
        final Client client = new Client("127.0.0.1", 8081);
        final RpcProtocol<RpcRequest> rpcProtocol = new RpcProtocol<>();

        // 构建消息头
        MsgHeader header = new MsgHeader();
        header.setMagic(ProtocolConstants.MAGIC);
        header.setVersion(ProtocolConstants.VERSION);
        header.setMsgType((byte) MsgType.REQUEST.ordinal());
        long requestId = RpcRequestHolder.getRequestId();
        header.setRequestId(requestId);
        header.setStatus((byte) 0x1);

        final byte[] serialization = RpcSerialization.JSON.name.getBytes();
        header.setSerializationLen(serialization.length);
        header.setSerializations(serialization);
        rpcProtocol.setMsgHeader(header);

        // 构建请求消息
        final RpcRequest rpcRequest = new RpcRequest();
        final Class<HelloService> aClass = HelloService.class;
        rpcRequest.setClassName(aClass.getName());
        final Method method = aClass.getMethod("hello", String.class);
        rpcRequest.setMethodName(method.getName());
        rpcRequest.setMethodCode(method.hashCode());
        rpcRequest.setServiceVersion("1.0");
        rpcRequest.setParameterTypes(method.getParameterTypes()[0]);
        rpcRequest.setParameter("pupil~");
        rpcProtocol.setBody(rpcRequest);

        // 客户端发送消息
        client.sendRequest(rpcProtocol);
        // 异步监听器
        RpcFuture<RpcResponse> future = new RpcFuture<>(new DefaultPromise<>(new DefaultEventLoop()),3000);

        // 根据请求id绑定异步监听器
        RpcRequestHolder.REQUEST_MAP.put(requestId,future);

        // 获取返回的响应消息，设置超时时间的阻塞等待
        RpcResponse rpcResponse = future.getPromise().sync().get(future.getTimeOut(), TimeUnit.MILLISECONDS);
        // 打印响应消息
        System.out.println(rpcResponse);
    }
}
